package mr

import (
	"bufio"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"
)

// for sorting by key.
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

// Map functions return a slice of KeyValue.
type KeyValue struct {
	Key   string
	Value string
}

// var workerId string

type Lworker struct {
	workerId   string
	mapTask    func(string, string) []KeyValue
	reduceTask func(string, []string) string
	// timeStamp  time.Time
	mapfile    string
	reducefile string
}

// var wmu sync.Mutex

// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

// Map A file
func (w *Lworker) MapFile(filename string) bool {

	file, err := os.Open(filename)
	// fmt.Println(w.workerId, "opened ", filename)
	if err != nil {
		log.Fatalf("MapFile cannot open %v", filename)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("MapFile cannot read %v", content)
	}
	file.Close()
	// fmt.Println(w.workerId, "is trying to map ", filename, "?")
	intermediate := w.mapTask(filename, string(content))
	sort.Sort(ByKey(intermediate))
	// fmt.Println(w.workerId, "succeed in ", filename, "result is ", intermediate)
	// wmu.Lock()
	for _, value := range intermediate {
		oname := "mr-tmp-" + strconv.Itoa(ihash(value.Key)%10)
		file, _ := os.OpenFile(oname, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0777)
		file.WriteString(value.Key + " " + value.Value + "\n")
	}
	// wmu.Unlock()
	return true
}

// reduce A file
func (w *Lworker) ReduceFile(filename string) bool {
	//get the last letter in th filename
	file, err := os.Open(filename)
	if err != nil {
		log.Fatalf("Reduce File cannot open %v", filename)
	}
	defer file.Close()
	//make a list to storage keyvalue
	intermediate := []KeyValue{}
	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		line := scanner.Text()
		parts := strings.Fields(line)
		kv := KeyValue{
			Key:   parts[0],
			Value: parts[1],
		}
		intermediate = append(intermediate, kv)
	}

	sort.Sort(ByKey(intermediate))
	flen := len(filename)
	oname := "mr-out-" + filename[flen-1:flen]
	ofile, _ := os.Create(oname)
	i := 0
	for i < len(intermediate) {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)
		}
		output := w.reduceTask(intermediate[i].Key, values)

		// this is the correct format for each line of Reduce output.
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)
		i = j
	}
	return true
}

func (w *Lworker) IsHealth(args *Args, reply *Reply) error {
	reply.code = "200"
	return nil
}

func (w *Lworker) Exit(args *Args, reply *Reply) error {
	log.Fatal(args.RequestBody, " running overtime !! Exiting...")
	return nil
}

// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	// build server waiting for operate
	// Your worker implementation here.
	w := Lworker{}
	w.sendRegiRequest("register")
	// individual worker
	w.server(mapf, reducef)
	go w.sendHeartBeat()
	// go w.handleReceiveTask()
	for isMapNotDone() {
		w.mapPhase()
	}
	time.Sleep(1 * time.Second)
	for isReduceNotDone() {

		w.reducePhase()
	}

	// for allDone() {
	// 	w.mapPhase()
	// 	time.Sleep(2 * time.Second)
	// 	w.reducePhase()
	// }
}

func (w *Lworker) ReceiveMapTask(args *Args, reply *Reply) error {
	w.mapfile = args.RequestBody
	reply.code = "200"
	return nil
}
func (w *Lworker) ReceiveReduceTask(args *Args, reply *Reply) error {
	w.reducefile = args.RequestBody
	reply.code = "200"
	return nil
}

// send heartbeat to coordinator
func (w *Lworker) sendHeartBeat() {
	for {
		data := struct {
			WorkerId  string
			TimeStamp time.Time
		}{
			WorkerId:  w.workerId,
			TimeStamp: time.Now(),
		}
		jsonData, err := json.Marshal(data)
		if err != nil {
			log.Fatalf("Error marshalling to JSON: %v", err)
		}

		args := Args{
			RequestBody: string(jsonData),
		}
		reply := Reply{}
		call("Coordinator.HeartBeat", &args, &reply)
		if strings.EqualFold(reply.code, "200") {
			log.Fatal("send heart beat failed")
		}
		time.Sleep(1 * time.Second)
	}
}

// func allDone() bool {
// 	args := Args{}
// 	reply := Reply{}
// 	call("Coordinator.Finish", &args, &reply)
// 	if reply.ResponseBody == "" {
// 		return false
// 	} else {
// 		return reply.ResponseBody == "false"
// 	}
// }

func (w *Lworker) mapPhase() {

	// wmu.Lock()
	filename := askMapTask(w.workerId)
	// fmt.Println("worker:", w.workerId, "received a Map task ,filename:", filename)
	// wmu.Unlock()
	if filename != "" && w.MapFile(filename) {
		if w.sendMapFinSign(filename) {
			fmt.Println("send map fin sign false")
		}
		// fmt.Println(w.workerId, " has finished the MapTask in ", filename)
	}

}

func (w *Lworker) reducePhase() {
	// wmu.Lock()
	filename := askReduceTask(w.workerId)
	// wmu.Unlock()
	// fmt.Println(w.workerId, "received a Reduce task ,filename:", filename)
	if filename != "" && w.ReduceFile(filename) {

		// wmu.Lock()
		if w.sendReduceFinSign(filename) {
			fmt.Println("send reduce fin sign false")
		}
		// wmu.Unlock()
		// fmt.Println(w.workerId, " has finished the ReduceTask in ", filename)
	}
}

func (w *Lworker) sendMapFinSign(filename string) bool {
	data := map[string]interface{}{
		"workerId": w.workerId,
		"fileName": filename,
	}

	jsonData, err := json.Marshal(data)
	if err != nil {
		log.Fatalf("Error marshalling to JSON: %v", err)
	}

	args := Args{
		RequestBody: string(jsonData),
	}
	reply := Reply{}
	call("Coordinator.HandleMapFinSign", &args, &reply)
	return reply.ResponseBody != "200"
}

func (w *Lworker) sendReduceFinSign(filename string) bool {
	data := map[string]interface{}{
		"workerId": w.workerId,
		"fileName": filename,
	}

	jsonData, err := json.Marshal(data)
	if err != nil {
		log.Fatalf("Error marshalling to JSON: %v", err)
	}

	args := Args{
		RequestBody: string(jsonData),
	}
	reply := Reply{}
	call("Coordinator.HandleReduceFinSign", &args, &reply)
	return reply.ResponseBody != "200"

}

// ask a map task from cooridnator
func askMapTask(workerId string) string {
	args := Args{
		RequestBody: workerId,
	}
	reply := Reply{}
	call("Coordinator.ProvideMapTask", &args, &reply)
	if reply.ResponseBody != "" {
		return reply.ResponseBody
	}
	return ""
}

// ask a reduce task from coordinator
func askReduceTask(workerId string) string {
	args := Args{
		RequestBody: workerId,
	}
	reply := Reply{}
	call("Coordinator.ProvideReduceTask", &args, &reply)
	if reply.ResponseBody != "" {
		// fmt.Println(reply.ResponseBody)
		return reply.ResponseBody
	}
	return ""
}

func isMapNotDone() bool {
	args := Args{}
	reply := Reply{}
	call("Coordinator.MapFinished", &args, &reply)
	ret := reply.ResponseBody
	return ret == "false"
}

func isReduceNotDone() bool {
	args := Args{}
	reply := Reply{}
	call("Coordinator.ReduceFinished", &args, &reply)
	ret := reply.ResponseBody
	return ret == "false"
}

func (w *Lworker) sendRegiRequest(Type string) error {

	//register message
	args := Args{}
	args.RequestType = Type
	uuid := uuid.NewString()

	mapData := map[string]interface{}{
		"workerId": uuid,
	}
	workerId := uuid
	data, err := json.Marshal(mapData)
	if err != nil {
		fmt.Print("sendReRequest failed caused by mapData to json failed!! Reason is ", err)
	}
	args.RequestBody = string(data)
	//reply
	reply := Reply{}
	ok := call("Coordinator.RegisterWorker", &args, &reply)
	if ok {
		// fmt.Println("succeed in register in coordinator,workerId:", workerId)
	} else {
		fmt.Println("rpc calling from woorker to coordinator(method:sendRegiRequest) failed")
	}
	w.workerId = workerId
	return nil
}

func (w *Lworker) server(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
	// start a thread that listens for RPCs from coordinator.go\
	w.mapTask = mapf
	w.reduceTask = reducef
	// TODO solve this problem:rpc register,Lworker has no suitavle
	rpc.Register(w)

	rpc.HandleHTTP()
	sockname := workerSock() + w.workerId
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
	// fmt.Print("worker rpc register succeed ,wokerId:", w.workerId)
}

// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
		fmt.Println("coordinator has been stopped..")
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}

// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
func CallExample() {

	// declare an argument structure.
	args := Args{}

	// fill in the argument(s).

	// declare a reply structure.
	reply := Reply{}

	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.
	ok := call("Coordinator.Example", &args, &reply)
	if ok {
		// reply.Y should be 100.
		fmt.Printf("reply.Y %v\n", reply)
	} else {
		fmt.Printf("call failed!\n")
	}
}
